package com.shujia.MR.mapjoin;

import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.HashMap;

/**
 *
 */
public class MapJoinMapper extends Mapper<LongWritable, Text, Text, NullWritable> {

    HashMap<String, String> subject = new HashMap<>();

    /**
     * Called once at the beginning of the task.
     * 该方法在Task任务执行时，会被调用一次，可以在该方法中加载文件放置内存中
     */
    @Override
    protected void setup(Mapper<LongWritable, Text, Text, NullWritable>.Context context) throws IOException, InterruptedException {
        //        System.out.println("执行了setup...");

        URI[] cacheFiles = context.getCacheFiles();

        FileSystem fileSystem = FileSystem.get(context.getConfiguration());
        FSDataInputStream fsDataInputStream = fileSystem.open(new Path(cacheFiles[0]));

        BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(fsDataInputStream));

        String line;
        while ((line = bufferedReader.readLine()) != null) {
            // 1000001,语文
            String[] split = line.split(",");
            subject.put(split[0], split[1]);
        }

        bufferedReader.close();
        fsDataInputStream.close();
        fileSystem.close();

    }

    /**
     *  缓存文件如果在map方法中加载，那么没处理一行数据，都需要加载一次缓存文件，不够高效
     */
    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, NullWritable>.Context context) throws IOException, InterruptedException {
        // 1500100001,1000004,29
        String[] split = value.toString().split(",");
        String name = subject.get(split[1]);
        context.write(new Text(split[0] + "\t" + name + "\t" + split[2]),NullWritable.get());
    }

}
